package com.gis.aoitest.source;

import com.gis.aoitest.beans.SensorReading;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import ru.yandex.clickhouse.ClickHouseConnection;
import ru.yandex.clickhouse.ClickHouseDataSource;
import ru.yandex.clickhouse.ClickHouseStatement;
import ru.yandex.clickhouse.settings.ClickHouseProperties;
import ru.yandex.clickhouse.settings.ClickHouseQueryParam;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;

public class SourceTest05_ClickHouse {

    public static void main(String[] args) throws Exception {
//        ClickHouseProperties properties = new ClickHouseProperties();
//// set connection options - see more defined in ClickHouseConnectionSettings
//        String url = "jdbc:clickhouse://localhost:8123/leon";
//
//// set default request options - more in ClickHouseQueryParam
//        properties.setSessionId("default-session-id");
//        properties.setDatabase("leon");
//        properties.setHost("localhost");
//        properties.setUser("default");
//        properties.setPassword("password123");
//
//
//
//        ClickHouseDataSource dataSource = new ClickHouseDataSource(url, properties);
//        String sql = "select * from test";
//        Map<ClickHouseQueryParam, String> additionalDBParams = new HashMap<>();
//// set request options, which will override the default ones in ClickHouseProperties
//        additionalDBParams.put(ClickHouseQueryParam.SESSION_ID, "new-session-id");
//
//
//        ClickHouseConnection conn = dataSource.getConnection();
//        ClickHouseStatement stmt = conn.createStatement();
//        // 可有可无
////        ResultSet rs = stmt.executeQuery(sql, additionalDBParams);
//        ResultSet rs = stmt.executeQuery(sql);
//        while(rs.next()){
//            Long  id = rs.getLong(1);
//            String name =rs.getString(2);
//            System.out.println(id + ", " + name);
//        }

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 从自定义source读取数据
        DataStreamSource<String> dataStream = env.addSource(new MyClickHouse());

        dataStream.print("ClickHouse");

        env.execute();

    }


    public static class MyClickHouse implements SourceFunction<String>{

        boolean running = true;


        @Override
        public void run(SourceContext<String> ctx) throws Exception {

            ClickHouseProperties properties = new ClickHouseProperties();
            // set connection options - see more defined in ClickHouseConnectionSettings
            String url = "jdbc:clickhouse://localhost:8123/leon";

            // 读取clickhouse
            properties.setSessionId("default-session-id");
            properties.setDatabase("leon");
            properties.setHost("localhost");
            properties.setUser("default");
            properties.setPassword("password123");
            ClickHouseDataSource dataSource = new ClickHouseDataSource(url, properties);
            String sql = "select * from test";


            ClickHouseConnection conn = dataSource.getConnection();
            ClickHouseStatement stmt = conn.createStatement();
            // 可有可无
//        ResultSet rs = stmt.executeQuery(sql, additionalDBParams);
            ResultSet rs = stmt.executeQuery(sql);
//            while(running) {

                while(rs.next()){
                    Long  id = rs.getLong(1);
                    String name =rs.getString(2);
                    ctx.collect(id + ", " + name);
                }

//                cancel();
//            }

            conn.close();

        }

        @Override
        public void cancel() {
            running = false;
        }
    }
}
